package com.example.demo.config;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.Map;

/**
 * 本地任务
 * 对 1  2 3  4 递增数字进行累加求和任务
 *
 * @author 星夜十七
 */
public class LocalTopology {

    /**
     * 自定义的Spout需要继承
     */
    public static class MySpout extends BaseRichSpout {
        private Map map;
        private TopologyContext topologyContext;
        private SpoutOutputCollector collector;

        /**
         * 这个方法时初始化方法 ,在Spout组件初始化的时候只执行一次
         * 一般会在这个方法内部实现一些初始化的代码
         * 例子:如果需要实现对mysql的操作,需要使用连接池,那么连接池初始化的代码就可以放在open()里面
         *
         * @param map                  storm 的配置类 可以保存配置信息在storm中传递
         * @param topologyContext      topology的上下文
         * @param spoutOutputCollector 发射器负责发射数据
         */
        @Override
        public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
            this.map = map;
            this.topologyContext = topologyContext;
            this.collector = spoutOutputCollector;
        }

        int num = 0;

        /**
         * 这个方法会被框架一直调用,
         * spout负责在nextTuple中向外发射数据
         * 这个方法每执行一次就会向外发射一次数据
         */
        @Override
        public void nextTuple() {
            num++;
            System.out.println("Spout是:"+num);
            //发射的参数可以是多个
//             collector.emit(new Values(num),new Values(num));
            collector.emit(new Values(num));
        }

        /**
         * 声明输出字段
         * 在开始执行一次
         * @param outputFieldsDeclarer
         */
        @Override
        public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
            //上面 发射器发射的时候发射几个 这里就定义几个,根据顺序一一对应
            outputFieldsDeclarer.declare(new Fields("num"));
        }
    }

    public static class  SumBolt extends BaseRichBolt {
        private Map map;
        private TopologyContext topologyContext;
        private OutputCollector collector;

        /**
         * 初始化方法,只会执行一次
         * @param map
         * @param topologyContext
         * @param outputCollector
         */
        @Override
        public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
            this.map=map;
            this.topologyContext=topologyContext;
            this.collector=outputCollector;
        }

        int sum = 0;
        /**
         * 此方法执行多次  只要上一个组件发射一条数据,那么这个bolt就收到这个数据然后调用 excute方法去处理数据
         * @param tuple
         */
        @Override
        public void execute(Tuple tuple) {
            Integer num = tuple.getIntegerByField("num");
            //因为 tuple就是个List,所以也可以用下标来取  也可以通过脚本获取
//            Integer integer = tuple.getInteger(0);
            sum+=num;
            System.out.println("和为:"+sum);
            // 注意:如果有下一个Bolt接收继续处理的话,需要用发射器发射一下
//            collector.emit(new Values(sum));

        }

        /**
         * 注意:如果这个组件没有调用 emit方法向外发射数据,那这个方法就不需要实现
         * 如果这个组件向外发射了数据,那在方法内部必须声明字段 与 excute发送的字段按顺序对应
         * @param outputFieldsDeclarer
         */
        @Override
        public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

        }
    }


    public static void main(String[] args) {
        //把Spout和Bolt 组装成一个 Topology去执行
        TopologyBuilder topologyBuilder = new TopologyBuilder();
        //声明自己组件ID和程序类,然后找上游,上一个组件的id和下一个组件的id,没有就不填
        topologyBuilder.setSpout("spoutid",new MySpout());
        topologyBuilder.setBolt("mybolt",new SumBolt()).shuffleGrouping("spoutid");

        //创建本地集群,模拟Storm集群
        LocalCluster localCluster = new LocalCluster();
        String simpleName = LocalTopology.class.getSimpleName();
        Config config = new Config();
        localCluster.submitTopology(simpleName,config,topologyBuilder.createTopology());
    }

}
